package com.sunday.common.mq.rabbit.study.spring.study;


import brave.Tracing;
import brave.messaging.MessagingTracing;
import brave.spring.rabbit.SpringRabbitTracing;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.config.ContainerCustomizer;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.amqp.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.support.RetryTemplate;

/**
 * org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration
 * https://github.com/openzipkin/brave/blob/master/instrumentation/spring-rabbit/README.md
 * <p>
 * spring-amqp二个核心类RabbitAdmin和RabbitTemplate类
 * 1.RabbitAdmin类完成对Exchange，Queue，Binging的操作，在容器中管理了RabbitAdmin类的时候，可以对Exchange，Queue，Binging进行自动声明。
 * 2.RabbitTemplate类是发送和接收消息的工具类。
 * <p>
 * RabbitTemplate类是简化RabbitMQ访问的工具类（发送和接收消息）
 * 1.使用RabbitTemplate进行消息的发送。
 * 2.使用SimpleMessageListenerContainer类监听队列，进行消息的消费。
 */
@Slf4j
@Configuration
public class RabbitBraveConfiguration {

    @Value("${spring.application.name}")
    private String applicationName;

    /**
     * https://docs.spring.io/spring-amqp/docs/current/reference/html/#consumer-events
     * 带批注方法的消息转换
     * 在调用侦听器之前，管道中有两个转换步骤。 第一步使用 将传入的 Spring AMQP 转换为 Spring 消息传递。 调用目标方法时，如有必要，消息负载将转换为方法参数类型。MessageConverterMessageMessage
     * 第一步的默认值是处理对象转换的 Spring AMQP。 所有其他保留为 . 在下面的讨论中，我们称之为“消息转换器”。MessageConverterSimpleMessageConverter String java.io.Serializable byte[]
     * 第二步的默认转换器是 ，它委托给转换服务 （的实例）。 在下面的讨论中，我们称之为“方法参数转换器”。GenericMessageConverterDefaultFormattingConversionService
     * <p>
     * https://docs.spring.io/spring-amqp/docs/current/reference/html/#simple-message-converter
     * SimpleMessageConverter
     * 该策略的默认实现称为 。 这是实例使用的转换器（如果未显式配置替代项）。 它处理基于文本的内容、序列化的 Java 对象和字节数组。MessageConverterSimpleMessageConverterRabbitTemplate
     * <p>
     * https://docs.spring.io/spring-amqp/docs/current/reference/html/#serializer-message-converter
     * SerializerMessageConverter
     * 这个转换器与 类似，只是它可以配置其他 Spring 框架和转换实现。SimpleMessageConverterSerializerDeserializerapplication/x-java-serialized-object
     * <p>
     * https://docs.spring.io/spring-amqp/docs/current/reference/html/#json-message-converter
     * Jackson2JsonMessageConverter
     * 本节介绍如何使用 与 进行转换。 它包含以下部分：Jackson2JsonMessageConverterMessage
     * 转换为消息
     * 从消息转换
     */
//    @Bean
    public GsonJsonMessageConverter gsonJsonMessageConverter() {
        return new GsonJsonMessageConverter();
    }

    @Bean
    public Jackson2JsonMessageConverter jackson2JsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    /**
     * https://github.com/openzipkin/brave/blob/master/instrumentation/spring-rabbit/README.md
     */
    @Bean
    public SpringRabbitTracing springRabbitTracing(Tracing tracing) {
        return SpringRabbitTracing.newBuilder(MessagingTracing.create(tracing)).build();
    }

//    /**
//     * aop pull模式链路埋点 推荐方式
//     */
//    @Bean
//    public TracingAfterReceivePostProcessor tracingAfterReceivePostProcessor(SpringRabbitTracing springRabbitTracing) {
//        return new TracingAfterReceivePostProcessor(springRabbitTracing);
//    }

    /**
     * https://docs.spring.io/spring-amqp/docs/current/reference/html/#cf-pub-conf-ret
     * {@link RabbitAutoConfiguration.RabbitTemplateConfiguration#rabbitTemplate(RabbitTemplateConfigurer, ConnectionFactory, ObjectProvider)}
     * 发布商确认并返回
     * 通过将属性设置为“true”和“true”来支持已确认（具有相关性）和返回的消息。CachingConnectionFactory publisherConfirmType ConfirmType.CORRELATED publisherReturns
     * 设置这些选项后，工厂创建的实例将包装在 中，用于方便回调。 当获得这样的通道时，客户端可以向 . 该实现包含用于路由确认或返回到相应侦听器的逻辑。 以下各节将进一步介绍这些功能。ChannelPublisherCallbackChannelPublisherCallbackChannel.ListenerChannelPublisherCallbackChannel
     * 另请参阅相关发布者确认和返回以及作用域内操作。simplePublisherConfirms
     * 有关更多背景信息，请参阅 RabbitMQ 团队的博客文章，标题为“介绍发布者确认”。
     */
    @Bean
    public RabbitTemplate rabbitTemplate(RabbitTemplateConfigurer configurer, ConnectionFactory connectionFactory, ObjectProvider<RabbitTemplateCustomizer> customizers, SpringRabbitTracing springRabbitTracing) throws Exception {
        RabbitTemplate template = new RabbitTemplate();
        configurer.configure(template, connectionFactory);

        /**
         * Send Message Brave
         */
        springRabbitTracing.decorateRabbitTemplate(template);

//        template.addAfterReceivePostProcessors(new TracingAfterReceivePostProcessor(springRabbitTracing));

        customizers.orderedStream().forEach((customizer) -> customizer.customize(template));
        /**
         * publisher-confirm-type: correlated
         */
        template.setConfirmCallback((correlationData, ack, cause) -> {
            log.info("correlationData : {}", correlationData);
            log.info("ack : {}", ack);
            log.info("cause : {}", cause);
        });
        /**
         * publisher-returns: true
         *
         * 相关发布者确认并返回
         * 支持发布者的实现确认并返回。RabbitTemplateAmqpTemplate
         * 对于返回的消息，模板的属性必须设置为或必须为特定消息计算。 此功能需要将其属性设置为 （请参阅发布者确认并返回）。 返回由客户端通过调用注册 发送到客户端。
         * 回调必须实现以下方法：mandatory true mandatory-expression true CachingConnectionFactory publisherReturns true RabbitTemplate.ReturnsCallback setReturnsCallback(ReturnsCallback callback)
         * void returnedMessage(ReturnedMessage returned);
         *
         * 具有以下属性：ReturnedMessage
         * message- 返回的消息本身
         * replyCode- 指示退货原因的代码
         * replyText- 返回的文本原因 - 例如NO_ROUTE
         * exchange- 消息发送到的交易所
         * routingKey- 使用的路由密钥
         */
        template.setReturnsCallback(returned -> {
            log.info("exchange : {}", returned.getExchange());
            log.info("routingKey : {}", returned.getRoutingKey());
            log.info("message : {}", returned.getMessage());
            log.info("replyCode : {}", returned.getReplyCode());
            log.info("replyText : {}", returned.getReplyText());
        });

//        template.setRetryTemplate(retryTemplate(template));

        template.setMessageConverter(jackson2JsonMessageConverter());
//        template.setMessageConverter(gsonJsonMessageConverter());

        return template;
    }

//    /**
//     * 手动链路追踪 模拟
//     */
//    @Bean
//    public TracingReceivePostProcessor tracingReceivePostProcessor(SpringRabbitTracing springRabbitTracing, RabbitTemplate rabbitTemplate) {
//        return new TracingReceivePostProcessor(springRabbitTracing, rabbitTemplate);
//    }

    /**
     * 添加重试功能
     * 从版本 1.3 开始，您现在可以将 配置为使用 来帮助处理代理连接问题。
     * 有关完整信息，请参阅春季重试项目。
     * 以下只是一个使用指数退避策略和默认值的示例，该策略在将异常抛给调用方之前进行了三次尝试。
     * RabbitTemplateRetry TemplateSimpleRetry Policy
     * <p>
     * 这在一定程度上受到限制，因为重试上下文仅包含字段。
     * 对于更复杂的用例，您应该使用外部，以便可以通过上下文的属性将其他信息传达给 。
     * 以下示例演示如何执行此操作：
     * RecoveryCallback lastThrowable
     * RetryTemplate RecoveryCallback
     */
    @Bean
    public RetryTemplate retryTemplate(RabbitTemplate rabbitTemplate) throws Exception {
        RetryTemplate retryTemplate = new RetryTemplate();
        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
        backOffPolicy.setInitialInterval(500);
        backOffPolicy.setMultiplier(10.0);
        backOffPolicy.setMaxInterval(10000);
        retryTemplate.setBackOffPolicy(backOffPolicy);
//        retryTemplate.execute(new RetryCallback<Object, Exception>() {
//            @Override
//            public Object doWithRetry(RetryContext context) throws Exception {
//                context.setAttribute("message", context.getRetryCount());
//                log.info("RetryCallback {}", context);
//                rabbitTemplate.convertAndSend(RabbitConfiguration.EXCHANGE_NAME, RabbitConfiguration.ROUTING_KEY_NAME, context.getRetryCount());
//                return context.getRetryCount();
//            }
//        }, new RecoveryCallback<Object>() {
//            @Override
//            public Object recover(RetryContext context) throws Exception {
//                Object message = context.getAttribute("message");
//                Throwable t = context.getLastThrowable();
//                log.info("RecoveryCallback {}, message {}", context, message);
//                // Do something with message
//                return null;
//            }
//        });

        return retryTemplate;
    }

    /**
     * {@link RabbitAnnotationDrivenConfiguration#simpleRabbitListenerContainerFactory(SimpleRabbitListenerContainerFactoryConfigurer, ConnectionFactory, ObjectProvider)}
     * 命名受限 @Bean(name = "rabbitListenerContainerFactory") 请参照上面的类
     */
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
            SimpleRabbitListenerContainerFactoryConfigurer configurer,
            ConnectionFactory connectionFactory,
            ObjectProvider<ContainerCustomizer<SimpleMessageListenerContainer>> simpleContainerCustomizer,
            SpringRabbitTracing springRabbitTracing
//            , RetryTemplate retryTemplate
    ) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        /**
         * Consumer Listener Brave
         */
        springRabbitTracing.decorateRabbitListenerContainerFactory(factory);
        simpleContainerCustomizer.ifUnique(factory::setContainerCustomizer);
        /**
         * 预取默认值曾经是 1，这可能导致高效使用者的利用率不足。 从版本 2.0 开始，默认预取值现在为 250，这应该让使用者在最常见的场景中保持忙碌和 从而提高吞吐量。
         * 尽管如此，在某些情况下，预取值应该很低：
         * 对于大型消息，尤其是在处理速度较慢的情况下（消息可能会在客户端进程中增加大量内存）
         * 当需要严格的消息排序时（在这种情况下，预取值应设置回 1）
         * 其他特殊情况
         * 此外，对于低容量消息传递和多个使用者（包括单个侦听器容器实例中的并发），您可能希望减少预取，以便在使用者之间更均匀地分配消息。
         */
        factory.setPrefetchCount(1);

        factory.setMessageConverter(jackson2JsonMessageConverter());
//        factory.setMessageConverter(gsonJsonMessageConverter());

        /**
         * 从版本 2.0.6 开始，可以将 和 添加到侦听器容器工厂。
         * 它在发送回复时使用。
         * 当重试次数用尽时调用。
         * 您可以使用 从上下文中获取信息。
         * 以下示例演示如何执行此操作：RetryTemplate RecoveryCallback RecoveryCallback SendRetryContextAccessor
         */
//        factory.setRetryTemplate(retryTemplate);
//        factory.setReplyRecoveryCallback(ctx -> {
//            Message failed = SendRetryContextAccessor.getMessage(ctx);
//            Address replyTo = SendRetryContextAccessor.getAddress(ctx);
//            Throwable t = ctx.getLastThrowable();
//            log.info("failed : {}, replyTo : {}, t : {}", failed, replyTo, t.getMessage());
//            return null;
//        });

        factory.setConsumerTagStrategy(queue -> applicationName + "_" + queue);

        return factory;
    }


//    @Bean
//    public RabbitTemplate rabbitTemplate(SpringRabbitTracing springRabbitTracing, ConnectionFactory connectionFactory) {
//        RabbitTemplate rabbitTemplate = springRabbitTracing.newRabbitTemplate(connectionFactory);
//        rabbitTemplate.setMandatory(true);
//        return rabbitTemplate;
////        return springRabbitTracing.newRabbitTemplate(connectionFactory);
//    }

//    @Bean
//    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(SpringRabbitTracing springRabbitTracing, ConnectionFactory connectionFactory) {
//        return springRabbitTracing.newSimpleRabbitListenerContainerFactory(connectionFactory);
//    }

//    @Bean
//    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, Tracer tracer) {
//        RabbitTemplate template = new RabbitTemplate();
//        template.setBeforePublishPostProcessors(message -> {
//            Span span = tracer.currentSpan();
//            if (span != null) {
//                message.getMessageProperties().setHeader("X-B3-TraceId", span.context().traceIdString());
//                message.getMessageProperties().setHeader("X-B3-SpanId", span.context().spanIdString());
//            }
//            return message;
//        });
//        return template;
//    }

    @Bean
    public RabbitListenerErrorHandler rabbitListenerErrorHandler() {
        return (amqpMessage, message, exception) -> {
            log.info("Message : {}", amqpMessage);
            log.info("org.springframework.messaging.Message : {}", message);
            log.info("ListenerExecutionFailedException : {}", exception);
            return null;
        };
    }


    /**
     * https://docs.spring.io/spring-amqp/docs/current/reference/html/#receiving-batch
     *
     * @RabbitListener批处理 接收一批消息时，通常由容器执行去批处理，并且一次使用一条消息调用侦听器。
     * 从版本 2.2 开始，您可以将侦听器容器工厂和侦听器配置为在一次调用中接收整个批处理，只需设置工厂的属性，并使方法有效负载参数为 a 或 ：batchListenerListCollection
     * <p>
     * 同样从 2.2 版开始。一次接收一条批处理消息时，最后一条消息包含设置为 的布尔标头。 可以通过将布尔 last' 参数添加到侦听器方法来获取此标头。
     * 标头映射自 。 此外，填充了每个消息片段中的批处理大小。true @Header(AmqpHeaders.LAST_IN_BATCH) MessageProperties.isLastInBatch() AmqpHeaders.BATCH_SIZE
     * 此外，还向 中添加了一个新属性。 如果为 true，容器将创建一批消息，最多 ;如果经过没有新消息到达，则会传递部分批处理。 如果收到生产者创建的批处理，则会对其进行去批处理并将其添加到使用者端批处理中;因此，实际传送的消息数可能超过 ，这表示从代理接收的消息数。 当为真时必须为真;容器工厂将强制实施此要求。consumerBatchEnabledSimpleMessageListenerContainerbatchSizereceiveTimeoutbatchSizedeBatchingEnabledconsumerBatchEnabled
     */
    @Bean
    public SimpleRabbitListenerContainerFactory consumerBatchContainerFactory(
            SimpleRabbitListenerContainerFactoryConfigurer configurer,
            ConnectionFactory connectionFactory,
            ObjectProvider<ContainerCustomizer<SimpleMessageListenerContainer>> simpleContainerCustomizer,
            SpringRabbitTracing springRabbitTracing
    ) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        springRabbitTracing.decorateRabbitListenerContainerFactory(factory);
        simpleContainerCustomizer.ifUnique(factory::setContainerCustomizer);

        factory.setConsumerTagStrategy(queue -> applicationName + "_" + queue);

//        factory.setPrefetchCount(1);
        factory.setBatchSize(2);
        // 同时也将设置 setBatchListener(true);
        factory.setConsumerBatchEnabled(true);

//        factory.setConcurrentConsumers();
//        factory.setMaxConcurrentConsumers();

//        factory.setIdleEventInterval(60000L);

        return factory;
    }

}
